package org.jetlinks.pro.rule.engine.cluster.strategies;

import lombok.AllArgsConstructor;
import org.jetlinks.pro.rule.engine.cluster.SchedulerSelectorStrategy;
import org.jetlinks.rule.engine.api.scheduler.ScheduleJob;
import org.jetlinks.rule.engine.api.scheduler.Scheduler;
import org.jetlinks.rule.engine.api.scheduler.SchedulingRule;
import org.jetlinks.rule.engine.cluster.SchedulerRegistry;
import org.jetlinks.rule.engine.cluster.TaskSnapshotRepository;
import org.springframework.stereotype.Component;
import reactor.bool.BooleanUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;

import java.util.Comparator;
import java.util.Objects;

/**
 * 选择最小负载对调度器来执行规则
 *
 * @author zhouhao
 * @since 1.3
 */
@AllArgsConstructor
public class MinimumLoadSchedulerSelectorStrategy implements SchedulerSelectorStrategy {

    public static final String type = "min_load";

    private final SchedulerRegistry registry;
    private final TaskSnapshotRepository repository;

    public static SchedulingRule newRule() {
        SchedulingRule schedulingRule = new SchedulingRule();

        schedulingRule.setType(type);

        return schedulingRule;
    }

    @Override
    public String getType() {
        return type;
    }

    @Override
    public Flux<Scheduler> select(Flux<Scheduler> schedulers, ScheduleJob job) {
        return schedulers
            .flatMap(scheduler -> scheduler.totalTask().zipWith(Mono.just(scheduler)))
            .sort(Comparator.comparing(Tuple2::getT1))
            .take(1)
            .map(Tuple2::getT2)
            ;
    }

    @Override
    public Mono<Boolean> test(Scheduler scheduler, ScheduleJob job) {
        //如果已经有其他调度器在调度任务,则不能再进行调度了
        return BooleanUtils
            .and(
                registry
                    .getSchedulers()
                    .filter(_scheduler -> !Objects.equals(_scheduler.getId(), scheduler.getId()))
                    .flatMap(_scheduler -> _scheduler
                        .getSchedulingTask(job.getInstanceId())
                        //存在调度中的任务
                        .map(task -> Objects.equals(task.getJob().getNodeId(), job.getNodeId()))
                    )
                    .all(Boolean.FALSE::equals),

                repository
                    .findByInstanceIdAndNodeId(job.getInstanceId(), job.getNodeId())
                    .hasElements()
                    .as(BooleanUtils::not)

            );
    }
}
